package com.yeming.flink.practice.watermark

import java.text.SimpleDateFormat

import com.yeming.flink.practice.source.StationLog
import org.apache.flink.api.common.functions.{AggregateFunction, ReduceFunction}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 每隔五秒计算一下最近十秒内，每个基站中通话时长最长的一次通话发生的时间，
 * 还有主叫号码，被叫号码，通话时长，并且还告诉我们当前发生的时间范围
 * 迟到五秒的数据不会丢弃，会放入到侧流中
 * watermark延迟2秒，允许的最大的延迟为5秒
 */
object TestWaterMark3 {

  def main(args: Array[String]): Unit = {

    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    streamEnv.setParallelism(1)
    streamEnv.getConfig.setAutoWatermarkInterval(100L) //周期性地引入watermark设置，默认的时间间隔就是100毫秒
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("localhost", 9999).map(line => {
      val arr: Array[String] = line.split(",")
      new StationLog(arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim, arr(4).trim.toLong, arr(5).trim.toLong)
    })
    //引入无序的数据，并且通过观察延迟的时间是3秒，采用周期性的watermark引入
      //代码写法又两种
      //第一种
//      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(3)) {
//        //抽取时间
//        override def extractTimestamp(element: StationLog): Long = {
//          element.callTime
//        }
    //第二种写法
      .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[StationLog] {
        var maxEventTime:Long = _
        override def getCurrentWatermark: Watermark = {
          new Watermark(maxEventTime - 2000L)
        }

        override def extractTimestamp(element: StationLog, previousElementTimestamp: Long): Long = {
          maxEventTime = maxEventTime.max(element.callTime)
          element.callTime
        }
      })

    val lateTag = new OutputTag[StationLog]("late")

    val result: DataStream[String] = stream.filter(_.callType.equals("success"))
      .keyBy(_.sid)
      .timeWindow(Time.seconds(10), Time.seconds(5))
      .allowedLateness(Time.seconds(5))
      .sideOutputLateData(lateTag)
      .aggregate(new MyAggregateCountFunction, new MyOutputResultWindowFunction)

    val lateStream: DataStream[StationLog] = result.getSideOutput(lateTag)
    lateStream.print("late")

    result.print("main")
    streamEnv.execute("TestWatermark")
  }

  class MyAggregateCountFunction extends AggregateFunction[StationLog,Long,Long]{

    override def createAccumulator(): Long = 0
    override def add(value: StationLog, accumulator: Long): Long = accumulator + 1

    override def getResult(accumulator: Long): Long = accumulator

    override def merge(a: Long, b: Long): Long = a + b
  }

  class MyOutputResultWindowFunction extends WindowFunction[Long,String,String,TimeWindow] {
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[String]): Unit = {
      val value: Long = input.iterator.next()
      val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      val str: String = String.format("当前窗口的时间范围是：%s-%s，当前呼叫的基站是：%s，当前基站呼叫的次数是：%d。",
        format.format(window.getStart), format.format(window.getEnd), key, Long.box(value))
      out.collect(str)
    }
  }
}
